#include "ControllerConnection.h"
#include "ScriptParser.h"
#include "Settings.h"
#include "StorageInterface.h"
#include "ReservedJobs.h"

bool Renderer::g_bClosed = false;
std::vector<Renderer*> Renderer::g_vecRenderer;
SocketWorker * Renderer::g_worker = NULL;


void Renderer::Finish()
{
	for(int i = 0 ; i < g_vecRenderer.size() ; i ++)
	{
		g_vecRenderer[i]->Stop();
		delete g_vecRenderer[i];
	}
}


int Renderer::Dist(const _JOB_INFO & job_info)
{
	int size = g_vecRenderer.size();
	int index = -1;
	for(int i = 0 ; i < size ; i ++)
	{
		if(!g_vecRenderer[i]->isWorking())
		{
			index = i;
			break;
		}
	}

	if(index == -1)
	{
		Renderer * nr = new Renderer;
		nr->activate();

		g_vecRenderer.push_back(nr);
		index = size;

		printf("[CREATE renderer thread (%d)]\n" , g_vecRenderer.size());
	}
	else
	{
		printf("[DIST to %dth renderer thread (%d)]\n" , index, g_vecRenderer.size());
	}

	Renderer* r = g_vecRenderer[index];

	r->job_info = job_info;

	ACE_Message_Block * mb = NULL;
	ACE_NEW_RETURN(mb , ACE_Message_Block(1) , -1); 
	mb->copy("0" , 1);
	r->putq(mb);


	return 0;
}

bool Renderer::isWorking()
{
	bool ret = false;

	renderer_lock.acquire();

	ret = job_info.IsValid();

	renderer_lock.release();

	return ret;

}

void Renderer::Clear()
{
	renderer_lock.acquire();

	job_info.Clear();

	renderer_lock.release();
}

int Renderer::Stop()
{
	stop = true;
	this->Clear();

	ACE_Message_Block * mb = NULL;
	ACE_NEW_RETURN(mb , ACE_Message_Block(1) , -1); 
	mb->copy("0" , 1);
	this->putq(mb);

	this->wait();
	return 0;
}

int Renderer::svc()
{
	ACE_Message_Block * mb = NULL;

	while(!stop)
	{
		if(this->getq(mb , NULL) != -1)
		{
			mb->release();
			Work();
		}
	}

	printf("Renderer Thread finished\n");

	return 0;
}



void Renderer::Work()
{
	if(!this->isWorking())
		return;

	std::string orgFileString = std::string("[Source : ") + job_info.GetSrcVideoFilePath(true) + " , " + job_info.GetSrcScriptFilePath(true) + "] ";

	time_t begin = time(NULL);

	LOG::PushJobLog(_INFO , job_info.GetJobId() , "started");
	JOB_MONITOR::StartJob(job_info);

	std::vector<_PROCESS_CMD> vecCmd;

	std::string filterId , effectIdList , soundId;
	
	_WORK_RESULT result;
	try{
		result = ScriptParser::MakeFFMpegCmd(job_info , vecCmd , filterId , effectIdList , soundId);
	}
	catch(std::runtime_error& re)
	{
		result.Set(WORK_RESULT_ERR_SCRIPT_STRUCTURE , std::string("invalid script structure : ") + re.what()); 
	}

	PACKET_WORK_END we(job_info.GetJobId());

	int cmd_count = vecCmd.size();

	if(!result.IsSuccess())
	{
		LOG::PushJobLog(_ERROR , job_info.GetJobId() ,orgFileString + result.msg); 
		printf("%s\n" , result.msg.c_str());

		PROCESS::DeleteTempFiles(job_info.GetJobId());
		JOB_MONITOR::EndJob(job_info , result);
		this->Clear();
		int size = g_worker->Send(we);

		return;
	}
	else if(cmd_count == 0)
	{
		result.Set(WORK_RESULT_ERR_SCRIPT_CONTENT, orgFileString + "invalid script content : no command");
		LOG::PushJobLog(_ERROR , job_info.GetJobId() ,result.msg);

		PROCESS::DeleteTempFiles(job_info.GetJobId());
		JOB_MONITOR::EndJob(job_info , result);
		this->Clear();
		int size = g_worker->Send(we);

		return;
	}

	for(int c = 0; c < vecCmd.size() && !g_bClosed ; c++)
	{
		_PROCESS_CMD & pc = vecCmd[c];

		if(Settings::IsDebugMode())
		{
			printf("cmd : %s\n" , pc.cmd.c_str());
			getchar(); 
		}

		_RESULT_TYPE ret = PROCESS::Execute(pc.type , pc.cmd);

		if(g_bClosed)
		{
			return;
		}

		if(_INVALID_CMD == ret)
		{
			result.Set(WORK_RESULT_ERR_RENDERING_PARAM , orgFileString + "invalid command :" + pc.cmd);
			LOG::PushJobLog(_ERROR , job_info.GetJobId() , result.msg);

			printf("%s\n" , result.msg.c_str());

			break;
		}
		else if(_FAIL == ret)
		{
			result.Set(WORK_RESULT_ERR_RENDERING_EXE , orgFileString + "execution failed  :" + pc.cmd);
			LOG::PushJobLog(_ERROR , job_info.GetJobId() , result.msg);

			printf("%s\n" , result.msg.c_str());

			break;
		}


	}

	if(result.IsSuccess())
	{
		LOG::PushObjectLog(job_info.GetJobId() , filterId , effectIdList , soundId);

		if(!StorageInterface::CopyOutput(job_info))
		{
			result.Set(WORK_RESULT_ERR_RENDERING_EXE , orgFileString + "copy output failed");
			LOG::PushJobLog(_ERROR , job_info.GetJobId() , result.msg);

			printf("%s\n" , result.msg.c_str());
		}
		else
		{
			time_t end = time(NULL);
			printf("*********** work[%s] elapsed : %ds ************\n",job_info.GetJobId().c_str() , (int)(end-begin) );

			LOG::PushJobLog(_REQUIRED , job_info.GetJobId() , job_info.GetDuration() , end-begin);
		}
	}

	PROCESS::DeleteTempFiles(job_info.GetJobId());
	JOB_MONITOR::EndJob(job_info , result);
	this->Clear();
	int size = g_worker->Send(we);

}



int ControllerWorker::handle_close (ACE_HANDLE h, ACE_Reactor_Mask mask)
{
	Renderer::g_bClosed = true;

	if(!timeout && 'k' != Settings::GetSettings()->OPERATION_MODE && 'K' != Settings::GetSettings()->OPERATION_MODE)
	{
		LOG::PushProcessLog(_ERROR , GetRemoteAddr() + " : controller connection closed abnormally");
	}

	ACE_Reactor::instance ()->end_event_loop();

	return 0;
}

bool ControllerWorker::Start(ACE_Reactor * reactor)
{
	ACE_SOCK_Stream stream;
	ACE_INET_Addr addr(Settings::GetSettings()->CONTROLLER_PORT , Settings::GetSettings()->CONTROLLER_ADDR.c_str());
	ACE_SOCK_Connector  connector;
	ACE_Time_Value timeout(10); 
	if(connector.connect(stream , addr , &timeout) == -1)
	{
		LOG::PushProcessLog(_ERROR , std::string("controller connection error : ") + Settings::GetSettings()->CONTROLLER_ADDR);
		printf("controller connection error=> %s:%d\n" ,Settings::GetSettings()->CONTROLLER_ADDR.c_str() , Settings::GetSettings()->CONTROLLER_PORT );
		return false;
	}

	ControllerWorker * pWorker = NULL;
	ACE_NEW_RETURN(pWorker , ControllerWorker , false);	

	pWorker->SetStream(reactor , Settings::GetSettings()->CONTROLLER_ADDR , stream); 

    if (reactor->register_handler(pWorker , ACE_Event_Handler::READ_MASK) == -1)
	{
		LOG::PushProcessLog(_ERROR , "can not register reactor");
		printf("can not register reactor");
		stream.close();
		delete pWorker;
		return false;
	}

	printf("connected to controller\n");
	LOG::PushProcessLog(_INFO, "connected to controller");

	Renderer::g_worker = (SocketWorker*)pWorker;
	
	PACKET_AGENT p;
	pWorker->Send(p);

	printf("[agent %d] send PID 2 controller\n" , getpid());

	return true;

}




bool ControllerWorker::Execute(const std::string &rcv_text)
{
	Json::Value root;
	if(!PACKET_BASE::GetRoot(rcv_text , root))
	{
		root.clear();
		return false;
	}

	std::string sCmd = PACKET_BASE::GetStringS(root , "CMD");

	ACE_ASSERT(sCmd.length() == 4);

	std::string renderer_id = PACKET_BASE::GetStringS(root , "RENDERER_ID");

	if(Settings::GetSettings()->RENDERER_ID != renderer_id)
	{
		printf("[agent] Different ID this : %s , controller : %s => close connection\n" , Settings::GetSettings()->RENDERER_ID.c_str() , renderer_id.c_str());

		root.clear();
		m_Stream.close();

		return false;
	}

	if(strcmp(sCmd.c_str() , _KEY_READ_SETTINGS)==0)
	{
		if(Settings::ApplyChangedSettingValue())
		{
			LOG::SendSMS("server setting has been changed");

			printf("\n#################### Changed Setting Values ################\n");
			printf("%s" , Settings::GetSettings()->dump().c_str());
			printf("###############################################################\n");

			if('K' == Settings::GetSettings()->OPERATION_MODE || 'k' == Settings::GetSettings()->OPERATION_MODE)
			{
				printf("operation mode : kill process\n");
				ACE_Reactor::instance ()->end_event_loop();
				return false;
			}

		}

	}

	else if(strcmp(sCmd.c_str() , _KEY_TICK)==0)
	{

	}
	else if(strcmp(sCmd.c_str() , _KEY_WORK)==0)
	{
		PACKET_WORK w(root);
		printf("%s\n" , w.Dump().c_str());

		Renderer::Dist(w.GetJobInfo());

	}



	return true;
}

